package ioc

import (
	"gitee.com/youkelike/zlog"
	"gitee.com/youkelike/zmigrate/example/interactive/repository/dao"
	"gitee.com/youkelike/zmigrate/example/interactive/service"
	"gitee.com/youkelike/zmigrate/example/interactive/web"
	"gitee.com/youkelike/zmigrate/migrator/events"
	"gitee.com/youkelike/zmigrate/migrator/events/canal"
	"gitee.com/youkelike/zmigrate/migrator/scheduler"
	"gitee.com/youkelike/zmigrate/migrator/validator"
	"gitee.com/youkelike/zmigrate/pkg/ginx"
	"gitee.com/youkelike/zmigrate/pkg/gormx/connpool"
	"gitee.com/youkelike/zmigrate/pkg/saramax"
	"github.com/gin-gonic/gin"
	"github.com/spf13/viper"
)

// NewConsumers 所有的 Consumer 在这里注册（除了 canal 的消费者）
func NewConsumers(l zlog.Logger, src SrcDB, dst DstDB) []saramax.Consumer {
	// 这里，有多少张表，你就初始化多少个 FixDataConsumer
	// 注意不同消费者、生产者和消费者之间都不能共用同一个 Kafka 客户端，否则有并发问题

	// 数据修复消费者
	interactive, err := events.NewFixDataConsumer[dao.Interactive](InitKafka(), l, src.db, dst.db)
	if err != nil {
		panic(err)
	}
	userLikeBiz, err := events.NewFixDataConsumer[dao.UserLikeBiz](InitKafka(), l, src.db, dst.db)
	if err != nil {
		panic(err)
	}
	return []saramax.Consumer{
		interactive,
		userLikeBiz,
		// 更多的 Consumer 在这里注册
	}
}

func InitMigratorWeb(
	l zlog.Logger,
	src SrcDB,
	dst DstDB,
	pool *connpool.DoubleWritePool,
	// 测试双写用到
	svc service.InteractiveService,
) *ginx.Server {
	engine := gin.Default()

	sch := scheduler.NewScheduler(l, pool)

	validators := make([]validator.Validator, 0, 5)
	consumers := make([]events.CanalConsumer, 0, 5)

	// 在这里，有多少张表，你就初始化多少个 producerXxx、consumerXxx、validatorXxx

	// 发送校验消息的生产者，全量校验器、增量校验器共用
	producerInteractive := events.NewFixDataProducer[dao.Interactive](InitKafka())
	// canal 的消费者（增量校验器）
	consumerInteractive := canal.NewCanalConsumer[dao.Interactive](InitKafka(), l, src.db, dst.db, src.dbName, dst.dbName, producerInteractive)
	// 全量校验器
	validatorInteractive := validator.NewFullValidator[dao.Interactive](src.db, dst.db, "SRC", l, producerInteractive)
	validators = append(validators, validatorInteractive)
	consumers = append(consumers, consumerInteractive)

	producerUserLikeBiz := events.NewFixDataProducer[dao.UserLikeBiz](InitKafka())
	consumerUserLikeBiz := canal.NewCanalConsumer[dao.UserLikeBiz](InitKafka(), l, src.db, dst.db, src.dbName, dst.dbName, producerUserLikeBiz)
	validatorUserLikeBiz := validator.NewFullValidator[dao.UserLikeBiz](src.db, dst.db, "SRC", l, producerUserLikeBiz)
	validators = append(validators, validatorUserLikeBiz)
	consumers = append(consumers, consumerUserLikeBiz)

	sch.SetCanalConsumers(consumers)
	sch.SetFullValidators(validators)
	sch.RegisterRoutes(engine.Group("/migrator/"))

	intr := web.NewInteractiveHandler(svc, l)
	intr.RegisterRoutes(engine.Group("/test/"))

	addr := viper.GetString("migrator.web.addr")
	return &ginx.Server{
		Addr:   addr,
		Engine: engine,
	}
}
